package com.atguigu.gmall.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.KafkaUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

/**
 * @author caodan
 * @version 1.0
 * @date 2022-09-19 8:58
 * 独立访客计算
 */
public class DwdTrafficVisitorComputingApp {

    public static void main(String[] args) throws Exception {


        // 创建环境变量
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置并行度
        env.setParallelism(4);

        // 开启检查点相关配置
        String pageTopicName = "dwd_traffic_page_log";
        String groupId = "dwd_traffic_page_group";
        FlinkKafkaConsumer<String> flinkKafkaConsumer = KafkaUtil.getFlinkKafkaConsumer(pageTopicName, groupId);

        //输入数据源
        DataStreamSource<String> kafkaDataStream = env.addSource(flinkKafkaConsumer);

        //格式转换 jsonStr -> jsonObj
        SingleOutputStreamOperator<JSONObject> fixDataStream = kafkaDataStream.map(JSON::parseObject);

        //分组
        KeyedStream<JSONObject, String> keyedDataStream = fixDataStream.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        //使用状态变量,记录每个访客的最后一次访问日志,
        SingleOutputStreamOperator<JSONObject> filterDataStream = keyedDataStream.filter(
                new RichFilterFunction<JSONObject>() {

                    //定义状态变量
                    ValueState<String> valueState;

                    // 初始化状态变量
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> lastVisitDateState = new ValueStateDescriptor<>("lastVisitDateState", String.class);
                        lastVisitDateState.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        valueState = getRuntimeContext().getState(lastVisitDateState);
                    }

                    @Override
                    public boolean filter(JSONObject jsonObj) throws Exception {
                        // 获取last_page_id ,不为空则不是独立访问
                        JSONObject lastPageId = jsonObj.getJSONObject("last_page_id");
                        if (lastPageId != null) {
                            return false;
                        }
                        //进行状态变量校验
                        String lastVisitDate = valueState.value();
                        Long ts = jsonObj.getLong("ts");
                        String visitDate = DateFormatUtil.toDate(ts);
                        if (lastVisitDate != null && lastVisitDate.equals(visitDate)) {
                            return false;
                        } else {
                            valueState.update(visitDate);
                            return true;
                        }
                    }
                }
        );
        filterDataStream.print();
        // 输出到kafka 主题中
        FlinkKafkaProducer<String> uniqueVisitorDetail = KafkaUtil.getFlinkKafkaProducer("dwd_traffic_unique_visitor_detail");
        SingleOutputStreamOperator<String> streamOperator = filterDataStream.map(JSONAware::toJSONString);
        streamOperator.addSink(uniqueVisitorDetail);
        //执行
        env.execute();
    }
}
